package data.platform.stream.internal.query;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import data.platform.common.service.command.KeyValueRepository;
import data.platform.common.service.query.MetricTagQueryService;
import data.platform.common.service.query.WindowAggregateQueryService;
import data.platform.common.stream.aggregates.*;
import data.platform.common.util.DateUtil;
import data.platform.stream.internal.cache.RuleAggregateCache;
import data.platform.stream.internal.cache.RuleCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Service
@Slf4j
@RequiredArgsConstructor
public class WindowAggregateQueryServiceImpl implements WindowAggregateQueryService {

    final KeyValueRepository<String, byte[]> keyValueRepository;

    final MetricTagQueryService metricTagQueryService;

    final RuleAggregateCache ruleAggregateCache;

    final RuleCache ruleCache;

    private static final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public Flux<WindowAggregateItemVO> queryAggregateItems(String ruleId) {

        if (!ruleAggregateCache.getCache().containsKey(ruleId)) {
            return Flux.empty();
        } else {
            List<WindowAggregateItemVO> vos = new ArrayList<>();

            Map<TimeWindow, Map<String, Map<Integer, WindowAggregateItem>>> timeWindowMap = ruleAggregateCache.getCache().get(ruleId);
            timeWindowMap.forEach(((timeWindow, metricMap) -> {
                metricMap.forEach((metricValue, tagIdMap) -> {
                    tagIdMap.forEach((tagIdValue, windowAggregate) -> {
                        String tag = metricTagQueryService.getTagJson(tagIdValue).get();
                        WindowAggregateItemVO vo = new WindowAggregateItemVO();
                        try {
                            vo.setTag(objectMapper.readValue(tag, Map.class));
                        } catch (JsonProcessingException ex) {
                            log.error("", ex);
                        }
                        vo.setStartTime(timeWindow.getStart());
                        vo.setEndTime(timeWindow.getEnd());
                        vo.setMetric(metricValue);
                        vo.setAvg(windowAggregate.getAvg());
                        vo.setMax(windowAggregate.getMax());
                        vo.setMin(windowAggregate.getMin());
                        vo.setCount(windowAggregate.getCount());
                        vo.setSum(windowAggregate.getSum());
                        vo.setLatestValue(windowAggregate.getLatestValue());
                        vo.setLatestEventTime(DateUtil.getDateOfLocalDateTime(windowAggregate.getLatestEventTime()).getTime());

                        vos.add(vo);
                    });
                });
            }));
            return Flux.fromIterable(vos);
        }

    }

    @Override
    public Flux<WindowResultVO> queryWindowResult(String ruleId) {
        if (ruleAggregateCache.getCache().containsKey(ruleId)) {
            Rule rule = ruleCache.findRule(ruleId).get();
            List<WindowResultVO> vos = new ArrayList<>();
            Map<TimeWindow, Map<String, Map<Integer, WindowAggregateItem>>> timeWindowCache = ruleAggregateCache.getCache().get(ruleId);
            for (Map.Entry<TimeWindow, Map<String, Map<Integer, WindowAggregateItem>>> timeWindowMapEntry : timeWindowCache.entrySet()) {
                TimeWindow timeWindow = timeWindowMapEntry.getKey();
                Map<String, Map<Integer, WindowAggregateItem>> metricCache = timeWindowCache.get(timeWindow);
                Map<Integer, WindowResult> comparisonResult = rule.apply(metricCache);
                comparisonResult.forEach((tagId, result) -> {
                    String tagJson = metricTagQueryService.getTagJson(tagId).get();
                    WindowResultVO vo = new WindowResultVO();
                    vo.setStartTime(timeWindow.getStart());
                    vo.setEndTime(timeWindow.getEnd());
                    try {
                        vo.setTag(objectMapper.readValue(tagJson, Map.class));
                    } catch (JsonProcessingException ex) {
                        log.error("", ex);
                    }
                    vo.setMetricValue(result.getMetricValue());
                    vo.setHit(result.getHit().get());

                    vos.add(vo);
                });
            }

            return Flux.fromIterable(vos);
        } else {
            return Flux.empty();
        }
    }

}
